package io.demo.message.serial;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import java.io.ByteArrayOutputStream;

/**
 * Serialization based on Kryo
 *
 * @param <T> Any type
 * @author lijinting01
 */
public class KryoSerializedObject<T> extends SerializedObject<T> {

  private Kryo kryo;

  private ByteArrayOutputStream byteArrayOutputStream;

  private Output kryoOutput;

  KryoSerializedObject(T object) {
    super(object);
  }

  @Override
  protected void beforeSerialize() {
    kryo = new Kryo();
    byteArrayOutputStream = new ByteArrayOutputStream();
    kryoOutput = new Output(byteArrayOutputStream);
  }

  @Override
  protected byte[] doSerialization(T object) {
    kryo.writeClassAndObject(kryoOutput, object);
    kryoOutput.flush();
    return byteArrayOutputStream.toByteArray();
  }

  @Override
  public T reverse() {
    Input input = new Input(getByteArray());
    @SuppressWarnings("unchecked")
    T target = (T) kryo.readClassAndObject(input);
    return target;
  }
}
